package obs;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

public class ManySocketV1 {
//    sub=title1:ack
//    send=title1:go
//    info 里的message like: title2:see

    public static void main(String[] args) {

    }

    public static void debug(Object o) {
        String s = "" + Thread.currentThread().getName();

        System.out.println(s + " thread: " + o.toString());
    }
}

class ManySSV1 {
    public static ConcurrentHashMap<String,String> states = new ConcurrentHashMap<String,String>();
    public static CopyOnWriteArraySet<Socket> sockets = new CopyOnWriteArraySet<>();
    public static ConcurrentHashMap<String, CopyOnWriteArraySet<Socket>> r = new ConcurrentHashMap<>();
    //    本线程池用处理建立链接的的socket的坚挺请求，本服务最多接受三个用户同时在线服务
    public static ExecutorService es = Executors.newFixedThreadPool(3);
    public static BlockingQueue<String> info = new LinkedBlockingQueue();
    public static void main(String[] args) {
        // 模拟主节点发生事件
        new Thread(() -> {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String s = scanner.nextLine();
                try {
                    info.put(s);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        },"服务端管理员提交信息线程").start();
        //专门发送info里的信息
        new Thread(()->{
            while (true){
                try {

                    String message = info.take();
                    String[] ss = message.split(":");
                    states.put(ss[0],ss[1]);
                    if (r.containsKey(ss[0])){
                        for (Socket socket : r.get(ss[0])) {
                            OutputStream out = socket.getOutputStream();
                            ManySocketV1.debug("服务器将向主题"+ss[0]+"写内容"+ss[1]);
                            out.write((ss[1]+"\r\n").getBytes());
                            out.flush();
                        }
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },"服务端专写线程").start();
        try {
            ServerSocket serverSocket = new ServerSocket(7777);
            while (true) {
                Socket s = serverSocket.accept();
                sockets.add(s);
                es.submit(new ServerReadTask(s, r,info));

            }


        } catch (IOException e) {
            ManySocketV1.debug("服务器启动失败");
            e.printStackTrace();
        }

    }
}

class ManySCV1 {
    public static void main(String[] args) {
        try {
            Socket socket = new Socket("master", 7777);
            //链接成功回调函数参数有socket
            ManySocketV1.debug("链接成功");
            new Thread(() -> {
                try {
                    BufferedReader reader = new BufferedReader(
                            new InputStreamReader(
                                    socket.getInputStream(), StandardCharsets.UTF_8));
                    while (true) {
                        String s = reader.readLine();
                        ManySocketV1.debug("监听到的消息　" + s);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }, "客户端监听读线程").start();
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String s = scanner.nextLine();
                BufferedWriter writer = new BufferedWriter(
                        new OutputStreamWriter(
                                socket.getOutputStream(), StandardCharsets.UTF_8));
                if (s.contains("=")&& s.contains(":")){
                    writer.write(s);
                    writer.newLine();
                    writer.flush();
                }else {
                    ManySocketV1.debug("格式不符合 "+s);
                }
            }
            //开启监听函数子线程，不过该线程目前不会读到任何消息，因为还没有订阅主题
        } catch (IOException e) {
            //链接失败回调函数
            ManySocketV1.debug("连接失败");
        }
    }
}

class ServerReadTask implements Runnable {
    private Socket socket;
    private ConcurrentHashMap<String, CopyOnWriteArraySet<Socket>> r;
    private BlockingQueue<String> info;

    public ServerReadTask(Socket socket,
                   ConcurrentHashMap<String,CopyOnWriteArraySet<Socket>> r,
                          BlockingQueue<String> info) {
        this.socket = socket;
        this.r = r;
        this.info=info;
    }

    @Override
    public void run() {
        try {
            BufferedReader reader = new BufferedReader(
                    new InputStreamReader(
                            socket.getInputStream(), StandardCharsets.UTF_8));
            while (true){
                String s = reader.readLine();
                if (s == null) break;
                ManySocketV1.debug("服务器接受到一个请求内容是：　" + s);
                String[] l = s.split("=");
                String command = l[0];
                String[] ll = l[1].split(":");
                String title = ll[0];
                String content = ll[1];
                ManySocketV1.debug("服务器接受到一个请求内容是：　" + command+title+content);
                if (command.equals("sub")) {
                    if (!r.containsKey(title)) {
                        CopyOnWriteArraySet<Socket> o = new CopyOnWriteArraySet<>();
                        r.put(title, o);
                    }
                    r.get(title).add(socket);
                    ManySocketV1.debug("服务器接受到一个订阅"+title);
                }else if (command.equals("send")){
                    info.put(l[1]);

                }else {
                    ManySocketV1.debug("服务端目前不支持该命令"+s);
                }
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
class ConcurrentSocket{
    private Socket socket;
//    ReentrantReadWriterLock;
    private BufferedWriter writer;
    private BufferedReader reader;
    private ReentrantLock readlock = new ReentrantLock();
    private ReentrantLock writelock = new ReentrantLock();

    public ConcurrentSocket(Socket socket) throws IOException {
        this.socket = socket;
         writer = new BufferedWriter(
                new OutputStreamWriter(
                        socket.getOutputStream(), StandardCharsets.UTF_8));
         reader = new BufferedReader(
                new InputStreamReader(
                        socket.getInputStream(), StandardCharsets.UTF_8));
    }
    public void  writerLineAndFlush(String line){
        writelock.lock();
        try{
            writer.write(line);
            writer.newLine();
            writer.flush();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            writelock.unlock();
        }
    }
    public String readLine(){
        String s =null;
        readlock.lock();
        try{
            s = reader.readLine();


        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            readlock.unlock();
        }
        return s;
    }



}